package cn.migu.music.hangupsms.kafka;

import cn.migu.music.common.utils.LogUtils;
import cn.migu.music.hangupsms.common.config.HangUpSmsConfig;
import cn.migu.music.hangupsms.common.entity.BillInfo;
import cn.migu.music.hangupsms.service.BaseService;
import cn.migu.music.hangupsms.service.ViewVrbtSendMsgService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Optional;

/**
 * 观看视频彩铃满足时长的话单处理
 */
@Component
@Slf4j
public class ViewVrbtListener {
    @Autowired
    private HangUpSmsConfig        hangUpSmsConfig;
    @Autowired
    private BaseService            baseService;
    @Autowired
    private ViewVrbtSendMsgService viewVrbtSendMsgService;

    @KafkaListener(topics = { "${kafka.view-vrbt-call-list.topic}" })
    public void handleKafkaMessage(ConsumerRecord<?, ?> record) {
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()) {
            String content = (String) kafkaMessage.get();

            if (hangUpSmsConfig.isViewVrbtNeedHandleKafka()) {
                try {
                    BillInfo billInfo = baseService.getBillInfo(content);
                    viewVrbtSendMsgService.contentProcess(billInfo);
                } catch (Exception businessException) {
                    LogUtils.info(businessException, log, "线程池队列已满");
                }
            } else {
                if (hangUpSmsConfig.isViewVrbtNeedLog()) {
                    LogUtils.info(log, "view vrbt:接收到kafka消息:开关:%s,话单内容:%s", hangUpSmsConfig.isViewVrbtNeedHandleKafka(), content);
                }
            }
        }
    }
}
